package com.example.rocketmq.tran;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * 发送事务消息(不支持延迟批量)
 * 
 * 事务消息的三种状态
 * 提交状态：提交事务，允许消费者消费此消息
 * 回滚状态：回滚事务，代表该消息将被删除，不能被消费
 * 中间状态：需要检查消息队列来确定状态
 */
public class TranProducer {
    public static void main(String[] args) throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer();
        producer.setProducerGroup("producer_group");
        producer.setNamesrvAddr("192.168.228.101:9876");
        producer.start();
        // 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            // 执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                if (StringUtils.equals("TAGA", message.getTags())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (StringUtils.equals("TAGB", message.getTags())) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if (StringUtils.equals("TAGC", message.getTags())) {
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }

            // MQ进行消息事务状态的回查，为了防止消息堆积，默认是超15次检查后丢弃消息
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println(messageExt.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        
        
        Message message = new Message();
        message.setTopic("tran_topic");
        message.setTags("TAGA");
        message.setBody("TAGA".getBytes());

        TransactionSendResult result = producer.sendMessageInTransaction(message, null);
        System.out.println(result);

        message = new Message();
        message.setTopic("tran_topic");
        message.setTags("TAGB");
        message.setBody("TAGB".getBytes());

        result = producer.sendMessageInTransaction(message, null);
        System.out.println(result);

        message = new Message();
        message.setTopic("tran_topic");
        message.setTags("TAGC");
        message.setBody("TAGC".getBytes());

        result = producer.sendMessageInTransaction(message, null);
        System.out.println(result);
        
        
    }
}


